package cn.uncode.mq.client.consumer;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.uncode.mq.client.NettyClient;
import cn.uncode.mq.network.Message;
import cn.uncode.mq.network.Topic;
import cn.uncode.mq.network.TransferType;
import cn.uncode.mq.server.RequestHandler;
import cn.uncode.mq.util.DataUtils;

public class Consumer extends NettyClient {
	
	private final static Logger LOGGER = LoggerFactory.getLogger(Consumer.class);
	
	private static final Consumer INSTANCE = new Consumer();
	
	private Lock lock = new ReentrantLock(true);
	private Set<ConsumerSubscriber> subscribers = new HashSet<ConsumerSubscriber>();
	private Set<String> topics = new HashSet<String>();
	private ConsumerRunnable consumerRunnableThread;
	
	
	private Consumer(){}
	
//	public static Consumer getInstance() {  
//		return INSTANCE;  
//	}
	
	public static void connect(String host, int port){
		INSTANCE.openning(host, port);
	}
	
	public static void fetch(){
		if(INSTANCE.topics.size() > 0){
			fetch(new ArrayList<String>(INSTANCE.topics));
		}
	}
	
	public static List<Topic> fetch(String[] topics){
		return fetch(Arrays.asList(topics));
	}
	
	public static List<Topic> fetch(List<String> topics){
		List<Topic> rtTopics = null;
		if(topics != null && topics.size() > 0){
			List<Topic> topicList = new ArrayList<Topic>();
			for(String tp : INSTANCE.topics){
				Topic topic = new Topic();
				topic.setTopic(tp);
				topicList.add(topic);
			}
			Message request = Message.newRequestMessage();
			request.setReqHandlerType(RequestHandler.FETCH);
			request.setBody(DataUtils.serialize(topicList));
			Message response = INSTANCE.write(request);
			if (response.getType() == TransferType.EXCEPTION.value) {
				// 有异常
				LOGGER.error("Cuonsumer fetch message error");
			} else {
				rtTopics = (List<Topic>) DataUtils.deserialize(response.getBody());
				LOGGER.info("=>> Consumer fetch message:"+rtTopics.toString());
			}
			//通知订阅者
			if(rtTopics != null){
				INSTANCE.lock.lock();
				try {
					for(Topic topic : rtTopics){
						for (ConsumerSubscriber subscriber : INSTANCE.subscribers) {
							if(subscriber.subscribeToTopic() != null && subscriber.subscribeToTopic().contains(topic.getTopic())){
								subscriber.notify(topic);
							}
						}
					}
				} finally {
					INSTANCE.lock.unlock();
				}
			}
		}
		return rtTopics;
	}
	
	public static void close(){
		INSTANCE.closing();
	}
	
	public static void addSubscriber(ConsumerSubscriber subscriber){
		INSTANCE.lock.lock();
		try {
			if (subscriber != null){
				if(!INSTANCE.subscribers.contains(subscriber)){
					INSTANCE.subscribers.add(subscriber);
				}
				INSTANCE.topics.addAll(subscriber.subscribeToTopic());
			}
		} finally {
			INSTANCE.lock.unlock();
		}
	}
	
	public static void deleteSubscriber(ConsumerSubscriber subscriber) {
		INSTANCE.lock.lock();
		try {
			if (subscriber != null){
				if(INSTANCE.subscribers.contains(subscriber)){
					INSTANCE.subscribers.remove(subscriber);
				}
				INSTANCE.topics.removeAll(subscriber.subscribeToTopic());
			}
		} finally {
			INSTANCE.lock.unlock();
		}
	}
	
	public static void runningConsumerRunnable(String host, int port){
		if (INSTANCE.consumerRunnableThread != null) {
			INSTANCE.consumerRunnableThread.close();
		}
		INSTANCE.consumerRunnableThread = new ConsumerRunnable(host, port);
		INSTANCE.consumerRunnableThread.start();
	}
	
	
	

}
